package com.mjf.orderpay_detect

import java.net.URL
import java.util

import com.mjf.dim.{OrderEvent, OrderResult}
import org.apache.flink.cep.{PatternSelectFunction, PatternTimeoutFunction}
import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * CEP 实现订单超时检测
 */
object OrderTimeout {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val resource: URL = getClass.getResource("/OrderLog.csv")
    val inputStream: DataStream[String] = env.readTextFile(resource.getPath)

    val orderEventStream: DataStream[OrderEvent] = inputStream
      .map {
        line =>
          val dataArr: Array[String] = line.split(",")
          OrderEvent(dataArr(0).toLong, dataArr(1), dataArr(2), dataArr(3).toLong)
      }
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[OrderEvent](Time.seconds(3)) {
        override def extractTimestamp(element: OrderEvent): Long = element.eventTime * 1000L
      })

    // 定义一个要匹配事件序列的模式
    val orderPayPattern: Pattern[OrderEvent, OrderEvent] = Pattern
      .begin[OrderEvent]("create").where(_.eventType == "create") // 订单的create事件
      .followedBy("pay").where(_.eventType == "pay") // 订单的pay事件
      .within(Time.minutes(15))

    // 将pattern应用到按照orderId分组的数据流上
    val patternStream: PatternStream[OrderEvent] = CEP.pattern(orderEventStream.keyBy(_.orderId), orderPayPattern)

    // 定义一个用来标识超时事件的测输出流
    val orderTimeoutOutputTag: OutputTag[OrderResult] = new OutputTag[OrderResult]("order timeout")

    // 调用select方法，提取匹配事件和超时事件，分别进行转换处理
    val resultStream: DataStream[OrderResult] = patternStream
      .select(orderTimeoutOutputTag, new OrderTimeoutSelect(), new OrderPaySelect())

    resultStream.print("pay")
    resultStream.getSideOutput(orderTimeoutOutputTag).print("timeout")

    env.execute("OrderTimeout")

  }
}

// 自定义超时处理函数
class OrderTimeoutSelect() extends PatternTimeoutFunction[OrderEvent, OrderResult] {
  // l : 超时时间戳
  override def timeout(map: util.Map[String, util.List[OrderEvent]], l: Long): OrderResult = {
    val timeoutOrderId: Long = map.get("create").get(0).orderId

    OrderResult(timeoutOrderId, "timeout at " + l)
  }
}

// 自定义匹配处理函数
class OrderPaySelect() extends PatternSelectFunction[OrderEvent, OrderResult] {
  override def select(map: util.Map[String, util.List[OrderEvent]]): OrderResult = {
    val payOrderId: Long = map.get("pay").get(0).orderId
    OrderResult(payOrderId, "payed successful")
  }
}